#include "ORCL.h"

void run_forward_thread(struct forward_thread_params *params){
    // 这个线程做两件事情，首先是根据一定的偏移量，不断监听转发缓冲区的值
    // 然后根据转发的目标节点，使用memcpy将数据拷贝到对应区域，或者执行新的一轮拷贝
    printf("listen-forward thread %d start\n", params->tid);

    struct orcl_context* orcl_ctx = params->orcl_ctx;
    int tid = params->tid;

    // 用一个变量来保存forward rc缓冲区的发送位置
    int forward_rc_offset = 0;
    // 监听的位置
    int listen_dc_offset = 0;

    // 头部
    volatile struct orcl_message_header* header= NULL;

    // 记录当前收到的包的数量
    int recv_count = 0;

    // 死循环，第一个字节是'a'代表有东西传入
    while(1){
        // 监听的是tid号转发接收缓冲区
        header = (volatile struct orcl_message_header*)(orcl_ctx->dct_forward_buf[tid] + listen_dc_offset);

        while(header->magic != 'a'){
            // 监听不到就持续监听
        }

        // printf("监听到了: node_id=%d, offset=%d, size=%d\n", header->node_id, header->offset, header->size);
        recv_count++;

        // 查看之前的一位，看看是不是阻塞了
        int last_offset = (listen_dc_offset - 512 + FORWARD_DCT_BUF_SIZE) % FORWARD_DCT_BUF_SIZE;
        if(*((char *)(orcl_ctx->dct_forward_buf[tid]+last_offset)) == 'a'){
            printf("forward node block:%d,%d,%d!\n", tid, recv_count, listen_dc_offset);
        }

        int node_id = header->node_id;
        int offset = header->offset;
        int size = header->size;

        // 这里有两条路线，一个是目标就是本地，那么直接memcpy，如果目标不是本地，那么就需要执行新的发送
        if(header->node_id == orcl_ctx->node_id){
            // 本地
            memcpy((void *)(orcl_ctx->message_buffer + header->offset), (void *)(orcl_ctx->dct_forward_buf[tid] + listen_dc_offset), header->size);

            memset((void *)(orcl_ctx->dct_forward_buf[tid] + listen_dc_offset), 0, sizeof(char));

        }else{
            // 查看转发的组是不是错了
            int remote_group_id = node_id / orcl_ctx->group_size;
            if(remote_group_id != orcl_ctx->group_id){
                printf("forward message error\n");
            }

            // 转发
            // printf("执行拷贝\n");
            // 将数据拷贝到组内转发缓冲区
            memcpy((void *)(orcl_ctx->rc_forward_buf[tid] + forward_rc_offset), (void *)(orcl_ctx->dct_forward_buf[tid] + listen_dc_offset), header->size);

            // printf("清空缓冲\n");
            // 清空这个缓冲位置
            memset((void *)(orcl_ctx->dct_forward_buf[tid] + listen_dc_offset), 0, sizeof(char));
            
            // 准备执行发送流程，使用forward qp
            orcl_inner_group_rdma_forward(orcl_ctx, tid, node_id % orcl_ctx->group_size, forward_rc_offset, offset, size);

            // 组内缓冲区转发位置自增
            forward_rc_offset = (forward_rc_offset + 512) % FORWARD_RC_BUF_SIZE;   
        }

        // 监听的位置自增
        listen_dc_offset = (listen_dc_offset + 512) % FORWARD_DCT_BUF_SIZE;
    }
}